/*
 * Copyright (C) 2015 kmy-framework-base Project
 *               Author: Administrator
 *               Date: 2015年11月4日
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.zmq.ha.client.declaration;

import java.io.IOException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import com.zmq.ha.client.core.ChannelWrapper;

public class ConsumerDelegate implements Consumer {
	
	private static final Logger log = LoggerFactory.getLogger(ConsumerDeclaration.class);

	private final ChannelWrapper channelHandler;
	public final Consumer delegate;
	private volatile boolean closed;
	private volatile String lastMessageId = "";
	

	public ConsumerDelegate(ChannelWrapper channelHandler, Consumer delegate) {
		this.channelHandler = channelHandler;
		this.delegate = delegate;
	}

	@Override
	public boolean equals(Object other) {
		return delegate.equals(other);
	}

	@Override
	public void handleCancel(String consumerTag) throws IOException {
		delegate.handleCancel(consumerTag);
	}

	@Override
	public void handleCancelOk(String consumerTag) {
		delegate.handleCancelOk(consumerTag);
	}

	@Override
	public void handleConsumeOk(String consumerTag) {
		delegate.handleConsumeOk(consumerTag);
	}

	@Override
	public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
			throws IOException {
		if (closed)
			return;
		
		try {
			long deliveryTag = envelope.getDeliveryTag();
			
			if( lastMessageId.equals( properties.getMessageId())){
				log.info(properties.getMessageId()+" has done with "+new String(body)+" delivery tag : "+deliveryTag);
				//requeue如果为true，则重传，如果为false则MQ服务器自动删除，不重新传递
				channelHandler.delegate.basicReject(deliveryTag, false);
				return;
			}
			lastMessageId = properties.getMessageId();
			channelHandler.maxDeliveryTag = deliveryTag = deliveryTag + channelHandler.previousMaxDeliveryTag;
			
			delegate.handleDelivery(consumerTag,
					new Envelope(deliveryTag, envelope.isRedeliver(), envelope.getExchange(), envelope.getRoutingKey()),
					properties, body);
		} catch (Exception e) {
			log.error("Error occur when handleDelivery message", e);
		}
		
	}

	@Override
	public void handleRecoverOk(String consumerTag) {
		delegate.handleRecoverOk(consumerTag);
	}

	@Override
	public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
		delegate.handleShutdownSignal(consumerTag, sig);
	}

	@Override
	public int hashCode() {
		return delegate.hashCode();
	}

	@Override
	public String toString() {
		return delegate.toString();
	}

	public void close() {
		closed = true;
	}

	public void open() {
		closed = false;
	}

}
